package org.dataporter.io.jdbc;

import org.apache.log4j.Logger;
import org.dataporter.data.DataLogEntry;
import org.dataporter.data.DataUpdate;
import org.dataporter.data.DataUpdateType;
import org.dataporter.io.DataWriter;
import org.dataporter.io.EntitySink;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class JDBCDataWriter<Sink extends EntitySink<? extends Connection>> implements DataWriter<Sink> {

    private static final Logger logger = Logger.getLogger( JDBCDataWriter.class );

    private final Map<String, Connection> connections = new HashMap<String, Connection>();

    private Sink sink;

    public void init( Sink sink ) throws Exception {
        this.sink = sink;
    }

    public DataLogEntry write( DataUpdate object ) {
        Connection connection = connectionForEntityName( object.getEntityName() );

        if ( object.getUpdateType() == DataUpdateType.Insert ) {
            return insert( connection, object );
        }
        else if ( object.getUpdateType() == DataUpdateType.Update ) {
            return update( connection, object );
        }
        else if ( object.getUpdateType() == DataUpdateType.Delete ) {
            return delete( connection, object );
        }

        throw new UnsupportedOperationException( "Unsupported update type: " + object.getUpdateType() );
    }

    private DataLogEntry insert( Connection connection, DataUpdate object ) {
        List<String> columnNames = new ArrayList<String>();
        List<Object> columnValues = new ArrayList<Object>();
        populateColumnData( object, columnNames, columnValues );

        return executePreparedStatement( connection, insertQueryString( object, columnNames ), columnValues );
    }

    private DataLogEntry delete( Connection connection, DataUpdate object ) {
        List<Object> columnValues = new ArrayList<Object>();
        populatePrimaryKeyValues( object, columnValues );

        return executePreparedStatement( connection, deleteQueryString( object ), columnValues );
    }

    private DataLogEntry update( Connection connection, DataUpdate object ) {
        List<String> columnNames = new ArrayList<String>();
        List<Object> columnValues = new ArrayList<Object>();
        populateColumnData( object, columnNames, columnValues );
        populatePrimaryKeyValues( object, columnValues );

        return executePreparedStatement( connection, updateQueryString( object, columnNames ), columnValues );
    }

    private void populateColumnData( DataUpdate object, List<String> columnNames, List<Object> columnValues ) {
        columnNames.addAll( object.getChanges().keySet() );

        for ( String name : columnNames ) {
            Object value = object.getChanges().get( name );
            columnValues.add( value );
        }
    }

    private void populatePrimaryKeyValues( DataUpdate object, List<Object> columnValues ) {
        for ( String pk : object.getPrimaryKeyNames() ) {
            columnValues.add( object.getChanges().get( pk ) );
        }
    }

    private String insertQueryString( DataUpdate object, List<String> columnNames ) {
        StringBuilder queryString = new StringBuilder( "INSERT INTO " ).append( object.getEntityName() ).append( " (" );

        StringBuilder columnValues = new StringBuilder();

        int i = columnNames.size();
        for ( String columnName : columnNames ) {
            i--;
            queryString.append( ' ' ).append( columnName );
            columnValues.append( " ?" );

            if ( i != 0 ) {
                queryString.append( ',' );
                columnValues.append( ',' );
            }

        }

        queryString.append( ") VALUES (" ).append( columnValues ).append( ");" );

        return queryString.toString();
    }

    private String deleteQueryString( DataUpdate object ) {
        StringBuilder queryString = new StringBuilder( "DELETE FROM " ).append( object.getEntityName() ).append( whereClause( object ) );
        return queryString.toString();
    }

    private String whereClause( DataUpdate object ) {
        StringBuilder whereClause = new StringBuilder( " WHERE " );
        List<String> primaryKeyNames = object.getPrimaryKeyNames();
        int i = primaryKeyNames.size();

        for ( String pk : primaryKeyNames ) {
            whereClause.append( pk ).append( " = " ).append( "?" );
            i--;

            if ( i != 0 ) {
                whereClause.append( " AND " );
            }
        }

        return whereClause.toString();
    }

    private String updateQueryString( DataUpdate object, List<String> columnNames ) {
        StringBuilder queryString = new StringBuilder( "UPDATE " ).append( object.getEntityName() ).append( " SET " );
        int i = columnNames.size();

        for ( String columnName : columnNames ) {
            queryString.append( columnName ).append( " = " ).append( "?" );
            i--;

            if ( i != 0 ) {
                queryString.append( ", " );
            }
        }

        queryString.append( " " ).append( whereClause( object ) ).append( ";" );

        return queryString.toString();
    }

    private DataLogEntry executePreparedStatement( Connection connection, String query, List<Object> values ) {
        try {
            PreparedStatement statement = connection.prepareStatement( query );

            for (int i=0; i<values.size(); i++) {
                Object o = values.get( i );
                statement.setObject( i+1, o );
            }

            statement.execute();
            return new DataLogEntry( query );
        }
        catch ( SQLException e ) {
            logger.error( "Failed to execute prepared statement", e );
            return new DataLogEntry( "Failed to execute prepared statement with exception: " + e.getClass().getName() );
        }
    }

    private Connection connectionForEntityName( String entityName ) {
        Connection connection = connections.get( entityName );

        if ( connection == null ) {
            connection = sink.sinkForEntity( entityName );
            connections.put( entityName, connection );
        }

        return connection;
    }

    public void close() {
        for ( Connection c : connections.values() ) {
            try {
                if ( !c.isClosed() ) {
                    c.close();
                }
            }
            catch ( Exception e ) {
                logger.error( "Failed to correctly close connection", e );
            }
        }
    }
}
